package com.atguigu.day10;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class FlinkSQL02_DataStreamToTable_Agg {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取端口数据并将数据转换为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("hadoop102", 9999)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new WaterSensor(fields[0],
                            Long.parseLong(fields[1]),
                            Double.parseDouble(fields[2]));
                });

        //3.将流转换为动态表
        //3.1 获取表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //3.2 转换
        Table sensorTable = tableEnv.fromDataStream(waterSensorDS);

        //4.使用TableAPI执行查询 select id,count(id) cnt from t group by id;
//        Table resultTable = sensorTable
//                .groupBy("id")
//                .select("id,id.count");
        Table resultTable = sensorTable
                .groupBy($("id"))
                .aggregate($("id").count().as("cnt"))
                .select($("id"), $("cnt"));

        //5.将动态表转换为流
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(resultTable, Row.class);

        //6.打印输出
        tuple2DataStream.print();

        //7.启动
        env.execute();

    }

}
